from datetime import datetime, timedelta, date
from airflow import DAG

from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.telegram.operators.telegram import TelegramOperator
from scipy.stats import norm

import psycopg2
import pandas as pd
import numpy as np
import sqlalchemy
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from airflow.models.xcom import XCom

import pendulum

import requests
from bs4 import BeautifulSoup
import logging
import re
from PIL import Image
import io
import json
import sys

import telebot
import geopy.distance

LOCAL_TZ = pendulum.timezone("Europe/Moscow")
# ------------------------------------------------------------------------------
# креды для Telegram
# ------------------------------------------------------------------------------
# TG_BOT_TOKEN = '7182058516:AAF_Cd8CXft4tcEkfUmWR6kXzEOrGF7FE3w'
# TG_CHAT_ID = '-4185266150'
# ------------------------------------------------------------------------------
# Конфиг с параметрами
# ------------------------------------------------------------------------------
with open('config/config.json') as f:
    config_json = json.load(f)
    conn_string = config_json['db_local_string']
    lot_data_columns = config_json['lot_data_columns']
    search_params = config_json['search_params']
    TG_BOT_TOKEN = config_json['prod_tg_bot_token']
    TG_CHAT_ID = config_json['prod_tg_chat_id']
# ------------------------------------------------------------------------------
# креды для db
# ------------------------------------------------------------------------------
#conn_string =  'postgresql://airflow:airflow@46.46.134.119:5432/airflow'
# ------------------------------------------------------------------------------
# SQL
# ------------------------------------------------------------------------------
col_list = ['lot_id', 'initial_price', 'average_price', 'step',
       'application_end_dt', 'oks_num', 'square', 'address_formatted',
       'longitude', 'latitude', 'm_longitude', 'm_latitude', 'm_name',
       'm_description', 'days_to_period_application_end', 'dist_m',
       'm_short_description', 'm_edit_name', 'm_short_name', 'distance', 'tag']
q_base = """
select
    t1.*,
    case when t2.price_upper_border <= t1.initial_price then t1.initial_price*1.02 else t2.price_upper_border end as price_upper_border
from
    (
    select
        ld.lot_id,
        ld.initial_price,
        ld.step,
        ld.application_end_dt,
        ldi.oks_num[1] as oks_num,
        (ldi.square::float[])[1],
        ldi.address_formatted,
        ldi.address_json,
        ldi.longitude,
        ldi.latitude,
        split_part(cord_json->'Point'->>'pos',
        ' ',
        1)::numeric as m_longitude,
        split_part(cord_json->'Point'->>'pos',
        ' ',
        2)::numeric as m_latitude,
        cord_json->>'name' as m_name,
        cord_json->>'description' as m_description,
        (ld.application_end_dt - now()::date) as days_to_period_application_end
    from
        (
        select
            *,
            row_number() over(partition by lot_id
        order by
            id desc) as rn
        from
            etl.lot_data ld
        ) ld
    join (
        select
            *,
            row_number() over(partition by lot_id
        order by
            id desc) as rn
        from
            etl.lot_data_info ld
            where address_formatted is not null) ldi
            using(lot_id)
    where
        ld.rn = 1
        and ldi.rn = 1
        and ld.action_flg <> 'D'
        and ld.trades_section <> 'Аренда/безвозмездное пользование'
        and ld.full_descr not ilike '%%комната%%'
        and is_share = 0
        and address_formatted is not null
        and (ldi.square::float[])[1]>= 20
        and initial_price<20000000
        and ld.lot_id not in (
        select
            lot_id
        from
            etl.lot_messages_to_tg)) t1
            join (select
    lot_id,
    price_upper_border
from
    (
    select
        lot_id,
        price_upper_border,
        row_number() over(partition by lot_id
    order by
        created_dttm desc) rn,
        created_dttm
    from
        etl.lot_data_dom_rf) t
where
    rn = 1) t2 on t1.lot_id = t2.lot_id
where
    days_to_period_application_end > 0
    and address_formatted not in ('Россия, Москва, Клинская улица, 2Ак7')
    and (address_json->'metaDataProperty'->'GeocoderMetaData'->'AddressDetails'->'Country'->'AdministrativeArea'->'Locality'->'Thoroughfare'->>'ThoroughfareName') 
    not in ('Береговой проезд')
order by
    days_to_period_application_end asc,
    initial_price asc;
"""


q_lot_media = """
select
    lot_id,
    media_link,
    width,
    height,
    position
from
    (
    select
        lot_id,
        media_link,
        width,
        height,
        position,
        max(case when media_md5 in ('aa602fb8365ae34d2fd3c70545e2c570', '885b142683479086f194fe940609901f') then 1 else 0 end) over(partition by lot_id) as aktarus_flg
    from
        etl.lot_media
    where
        loaded_flg = 1
        and media_md5 not in ('45bb0f3e0338fc5703f3b69891047c84', 'bf2a38e05f61d6bfc14dcf9907a1d8a4')
        and (width + height)>= 1000) t
where
    t.aktarus_flg = 0;
"""

q_lot_media_2gis = """
select
    lot_id,
    media_link,
    width,
    height,
    position
from
    etl.lot_media_2gis
where
    loaded_flg = 1;
"""

q_rnv = """
select address_json from etl.renovation_objects
"""

q_mos_obj = """
select
    mo.*,
    moa.address_json
from
    etl.mos_objects mo
join etl.mos_objects_address moa on
    replace(mo.house_url,
    'http://address.mosopen.ru/',
    '') = moa.id
"""


# ------------------------------------------------------------------------------
# маска рассылки
# ------------------------------------------------------------------------------

message_text = """
❗️Дисконт {sale_pct}%❗️\n

🏬{address_url}: {address}\n
📏Площадь: {square} кв.м\n
{distance_row}\n
{rnv_row}
-------------------------------
❗️Начальная цена - <tg-spoiler>{initial_price} млн.₽</tg-spoiler>\n
❗️Рыночная цена - {average_price} млн.₽\n
🗓Окончание приема заявок: {application_end_dt}
-------------------------------
☎️Подробнее об объекте и вариантах его приобретения:\n
+7(925)952-75-83\n
Больше информации в нашей группе {group_link}
Аккаунт для связи: {contact_link}

{tag_row}
"""
def get_house_address(address_json):
    address_search = []
    _components = address_json['metaDataProperty']['GeocoderMetaData']['Address']['Components']
    for _ in _components:
        if _['name'] not in ['Россия', 'Центральный федеральный округ']+address_search and \
        _['kind'] not in ['entrance', 'other']:
           address_search.append(_['name'])
    return address_search

def get_street_house(x):
    elements_list = []
    for element in x['metaDataProperty']['GeocoderMetaData']['Address']['Components']:
        if element['kind'] in ['street', 'house']:
            elements_list.append(element['name'])
    if len(elements_list) == 2:
        return ', '.join(elements_list)
    else:
        return None

def get_rnv_flg(street_house, rnv_street_house_list):
    if street_house != None:
        return int(street_house in rnv_street_house_list)
    else:
        return 0

def get_distance(cord1, cord2):
    return round(geopy.distance.geodesic(cord1, cord2).m)

def get_short_description(m_description):
    if 'мцд' in m_description.lower():
        return 'МЦД'
    elif 'московское центральное кольцо' in m_description.lower():
        return 'МЦК'
    else:
        return 'метро'

def get_distance_str(dist, coef = 1.2):
    corr_dist = dist * coef
    if corr_dist < 1000:
        dist = ((corr_dist // 50))*50
        return str(int(dist)) + ' метров' if int(dist) > 0 else '50 метров'
    else:
        return str(round(corr_dist/1e3, 1)) + ' км' if round(corr_dist/1e3, 1) > 1 else '1 км'

def create_link_location(lot_dict, word, url = "https://yandex.ru/maps/?whatshere[point]={longitude},{latitude}&whatshere[zoom]=17"):
    url = url.format(longitude = lot_dict['longitude'], latitude = lot_dict['latitude'])
    return """<a href='{url}'>{word}</a>""".format(url = url, word = word)

def create_link(url, word):
    return """<a href='{url}'>{word}</a>""".format(url = url, word = word)

# определение типа адреса
def get_address_type(components):
    components = components['metaDataProperty']['GeocoderMetaData']['Address']['Components']
    geo_dict = {'province':None,
                'locality':None,
               'street':None, 
               'house':None,
               'area':None}
    # вытаскиваем улицу и дом
    
    for component in components:
        if component['kind'] in ['street', 'house', 'province', 'locality']:
            geo_dict[component['kind']] = component['name']
        if component['kind'] == 'area':
            # в этом случае не ищем соответствий
            if component['name'] == 'Новомосковский административный округ':
                geo_dict['area'] = 'Новомосковский административный округ'
            elif component['name'] == 'Троицкий административный округ':
                geo_dict['area'] = 'Троицкий административный округ'
                

    if geo_dict['province'] == 'Москва' and geo_dict['locality'] == 'Москва' and geo_dict['street'] != None:
        return 1
    elif geo_dict['province'] == 'Москва' and geo_dict['locality'] == 'Зеленоград':
        return 2
    elif geo_dict['province'] == 'Москва' and geo_dict['locality'] == 'Москва' and geo_dict['street'] == None:
        return 3
    elif geo_dict['area'] == ['Новомосковский административный округ', 'Троицкий административный округ'][0]:
        return 4
    elif geo_dict['area'] == ['Новомосковский административный округ', 'Троицкий административный округ'][1]:
        return 5

def get_address_components(components, component_name):
    # шаблон
    geo_dict = {'province':None,
                'locality':None,
               'street':None, 
               'house':None}
    # вытаскиваем улицу и дом
    for component in components:
        if component['kind'] in ['street', 'house']:
            geo_dict[component['kind']] = component['name']
    return geo_dict[component_name]


def form_lot_card_from_pg(**kwargs):
    conn = create_engine(conn_string, connect_args={"options": "-c timezone=Europe/Moscow"})
    logging.info("""Start getting new lots""")
    df_base = pd.read_sql_query(sql = q_base, con = conn)
    df_rnv = pd.read_sql_query(sql = q_rnv, con = conn)
    # справочник адресов Москвы
    df_mos_objects = pd.read_sql_query(sql = q_mos_obj, con = conn)
    # берем улицу и дом 
    df_mos_objects['street'] = df_mos_objects.address_json.apply(lambda x:  get_address_components(x['metaDataProperty']['GeocoderMetaData']['Address']['Components'], component_name = 'street'))
    df_mos_objects['house'] = df_mos_objects.address_json.apply(lambda x: get_address_components(x['metaDataProperty']['GeocoderMetaData']['Address']['Components'], component_name = 'house'))
    df_mos_objects['longitude'] = df_mos_objects.address_json.apply(lambda x: x['Point']['pos'].split(' ')[0]).astype(float)
    df_mos_objects['latitude'] = df_mos_objects.address_json.apply(lambda x: x['Point']['pos'].split(' ')[1]).astype(float)
    df_mos_objects_dedup = df_mos_objects[['street', 'house', 'area', 'district','longitude', 'latitude']].drop_duplicates()
    # опрелеляем флаг реновации
    df_base['address_house'] = df_base.address_json.apply(lambda x: ', '.join(get_house_address(x)))
    df_base['street_house'] = df_base.address_json.apply(lambda x: get_street_house(x))
    df_rnv['street_house'] = df_rnv.address_json.apply(lambda x: get_street_house(x))
    df_base['rnv_flg'] = df_base.street_house.apply(lambda x: get_rnv_flg(x, [x for x in df_rnv.street_house.unique() if x!=None]))
    df_media, df_media_2gis =  pd.read_sql_query(sql = q_lot_media, con = conn),  pd.read_sql_query(sql = q_lot_media_2gis, con = conn)
    # нас интересует первый лот в списке
    for lot_id in df_base.lot_id.unique():
        try:
            print(lot_id)
            lot_df = df_base[df_base['lot_id']==lot_id].copy()
            # параметры для distance_row
            lot_df['dist_m'] = lot_df.apply(lambda x: get_distance((x['latitude'], x['longitude']), (x['m_latitude'], x['m_longitude'])), axis = 1)

            lot_df['m_short_description'] = lot_df['m_description'].apply(lambda x: get_short_description(x))

            lot_df['m_edit_name'] = lot_df['m_name'].apply(lambda x: x.replace('станция' , 'станции'))

            lot_df['m_short_name'] = lot_df['m_name'].apply(lambda x: x.replace('станция ' , '').replace('метро ', ''))

            lot_dict = {col:lot_df[col].values[0] for col in lot_df.columns}

            np.random.seed(int(round(datetime.now().timestamp())))

            lot_dict['average_price'] = lot_df['price_upper_border'].values[0]

            lot_dict['distance'] = get_distance_str(lot_dict['dist_m'])

            # позже обернуть в функцию!
            if lot_dict['dist_m']*1.2 > 2400:
                if lot_dict['m_short_description'] != 'метро' and 'метро' not in lot_dict['m_name'].lower():
                    distance_row = """🚇Ближайшая станция: {m_short_name} ({m_short_description})""".format(m_short_name = lot_dict['m_short_name'],
                                                                                                     m_short_description = lot_dict['m_short_description'])
                else: 
                    distance_row = """🚇Ближайшая станция: {m_name}""".format(m_name = lot_dict['m_name'])
            else:
                if lot_dict['m_short_description'] != 'метро' and 'метро' not in lot_dict['m_name'].lower():
                    distance_row = """🚇{distance} до {m_edit_name} ({m_short_description})""".format(distance = lot_dict['distance'],
                                                                                                    m_edit_name = lot_dict['m_edit_name'],
                                                                                                    m_short_description = lot_dict['m_short_description'])
                else:
                     distance_row = """🚇{distance} до {m_name}""".format(distance = lot_dict['distance'],
                                                                         m_name = lot_dict['m_name'])
            if lot_dict['rnv_flg'] == 1:
                rnv_row = """💎Объект участвует в программе <b>РЕНОВАЦИИ</b>💎\n"""
                rnv_tag = """#Реновация""" + ' '
            else:
                rnv_row = ''
                rnv_tag = ''

            # тэги
            lot_df['address_type'] = lot_df.address_json.apply(lambda x: get_address_type(x))
            components = lot_df.address_json.values[0]['metaDataProperty']['GeocoderMetaData']['Address']['Components']
            lot_cord = [lot_df.latitude.values[0], lot_df.longitude.values[0]]
            if lot_df['address_type'].values[0] == 1:
                street, house = get_address_components(components, component_name = 'street'), get_address_components(components, component_name = 'house')
                address_df = df_mos_objects_dedup[(df_mos_objects_dedup['street'] == street)&(df_mos_objects_dedup['house']==house)].copy()
                if len(address_df) == 0:
                    tmp_mos_objects = df_mos_objects_dedup.copy()
                    tmp_mos_objects['dist'] = tmp_mos_objects.apply(lambda x: get_distance([x.latitude, x.longitude], lot_cord), axis = 1)
                    top_20_obj = pd.DataFrame(tmp_mos_objects.sort_values(by = 'dist', ascending = True)[:20][['area', 'district']].value_counts())
                    area, district = top_20_obj.index[0][0], top_20_obj.index[0][1].replace(' ', '').replace('-', '')
                    area_tag = '#' + area.split(' ')[1][1:-1] + ' '
                    district_tag = '#' + district + ' '
                else:
                    area, district = address_df['area'].values[0], address_df['district'].values[0].replace(' ', '').replace('-', '')
                    area_tag, district_tag = '#' + area.split(' ')[1][1:-1] + ' ',  '#' + district + ' '
                    # позаимствуем отсюда геопозицию дома
                    lot_dict['longitude'] = address_df.longitude.values[0]
                    lot_dict['latitude'] = address_df.latitude.values[0]

            elif lot_df['address_type'].values[0] == 2:
                area = 'Зеленоградский (ЗелАО)'
                area_tag = '#'  + area.split(' ')[1][1:-1] + ' '
                district_tag = ''
            elif lot_df['address_type'].values[0] == 3:
                area_tag = ''
                district_tag = ''
            elif lot_df['address_type'].values[0] == 4:
                area = 'Новомосковский (НАО)'
                area_tag = '#' + area.split(' ')[1][1:-1] + ' '
                district_tag = ''
            elif lot_df['address_type'].values[0] == 5:
                area = 'Троицкий (ТАО)'
                area_tag = '#'  + area.split(' ')[1][1:-1] + ' '
                district_tag = ''


            
            # захардкодим теги
            tag_row = """{district_tag}{area_tag}{rnv_tag}#Москва #торги #недвижимость #квартиры""".format(area_tag = area_tag, district_tag = district_tag, rnv_tag = rnv_tag)
            lot_dict['tag'] = tag_row


            message_tg = message_text.format(address = lot_dict['address_house'], 
                                 address_url = create_link_location(lot_dict, word = 'Адрес'),
                                 square = lot_dict['square'],
                                 initial_price = round(lot_dict['initial_price']/1e6, 2),
                                 average_price = round(lot_dict['average_price']/1e6, 2),
                                 sale_pct = round(100*(1 - lot_dict['initial_price']/lot_dict['average_price'])),
                                 application_end_dt = date.strftime(lot_dict['application_end_dt'], '%d.%m.%Y'),
                                 group_link = create_link(url = "https://t.me/rostorgi_moskva", word = 'РосТорги Москва'),
                                 contact_link = create_link(url = "https://t.me/rostorgi_msk", word = 'РосТорги'),
                                 distance_row = distance_row,
                                 rnv_row = rnv_row,
                                 tag_row = tag_row)
            # фотки
            media_link_mask = 'https://xn--p1abd.xn----etbpba5admdlad.xn--p1ai/pictures/'
            media_link_mask_2gis = 'https://'
            # фотки с сайта торгов
            media_list = []
            media_links = df_media[df_media['lot_id'] == lot_dict['lot_id']].sort_values(by = 'position').media_link.values
            for media_link in media_links:
                url = media_link_mask + media_link
                img_data = requests.get(url).content
                img = Image.open(io.BytesIO(img_data))
                if (sys.getsizeof(img.tobytes())/1e6)<10:
                    media_list.append(telebot.types.InputMediaPhoto(img, parse_mode = 'HTML'))
                else:
                    # меняем размер
                    coef = (sys.getsizeof(img.tobytes())/1e6)/10
                    resize_img = img.resize([int(round(img.size[0]/coef, -2)), int(round(img.size[1]/coef, -2))])
                    width,height = resize_img.size
                    media_list.append(telebot.types.InputMediaPhoto(resize_img.crop((0,0,width,height-50)), parse_mode = 'HTML'))
            media_links_2gis = df_media_2gis[df_media_2gis['lot_id'] == lot_dict['lot_id']].sort_values(by = 'position').media_link.values
            # берем первые 2 фотки
            for media_link in media_links_2gis:
                url = media_link_mask_2gis + media_link
                img_data = requests.get(url).content
                img = Image.open(io.BytesIO(img_data))
                width,height = img.size
                if (sys.getsizeof(img.tobytes())/1e6)<10:
                    media_list.append(telebot.types.InputMediaPhoto(img.crop((0,0,width,height-50)), parse_mode = 'HTML'))
                else:
                    # меняем размер
                    coef = (sys.getsizeof(img.tobytes())/1e6)/10
                    resize_img = img.resize([int(round(img.size[0]/coef, -2)), int(round(img.size[1]/coef, -2))])
                    width,height = resize_img.size
                    media_list.append(telebot.types.InputMediaPhoto(resize_img.crop((0,0,width,height-50)), parse_mode = 'HTML'))
            if len(media_list) != 0:
                media_list[0].caption = message_tg
            if len(media_list) > 10:
                media_list = media_list[:10].copy()
            # отправка
            if len(media_list) > 0:
                try:
                    bot = telebot.TeleBot(TG_BOT_TOKEN)
                    bot.send_media_group(TG_CHAT_ID, media_list)
                    pd.DataFrame([lot_dict])[col_list].to_sql(name = 'lot_messages_to_tg', schema = 'etl', con = conn, if_exists='append', index = False)
                    break
                except Exception:
                    print('Send to bot finish with exception!')
                    break
        except Exception:
            print('Exception, trying next!')
            pass

with DAG(
        dag_id='new_lots_to_tg',
        start_date = datetime(year = 2023, month = 4, day = 23, hour = 8, minute = 0, tzinfo=LOCAL_TZ),
        schedule_interval='30 10-20/1 * * *',
        catchup=False,
        default_args = {
           'owner':'d.kakurin',
       },
        tags=['tg'],
) as dag:
    start_task = EmptyOperator(
        task_id="start_task"
    )

    get_data_from_pg = PythonOperator(
        task_id='get_data_from_pg',
        python_callable=form_lot_card_from_pg
    )

    # print_xcom = PythonOperator(
    #     task_id='printer',
    #     python_callable=print_xcom
    # )

    end_task = EmptyOperator(
        task_id="end_task"
    )
start_task >> get_data_from_pg >> end_task